package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.bridge.scala._

object OverTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    val click_table = tEnv.fromDataStream(clickStream,
      $("user"),
      $("url"),
      $("ts").rowtime())
    tEnv.createTemporaryView("click_table", click_table)

    // 统计1分钟内是否有过访问
    val result = tEnv.sqlQuery(
      s"""
         |select user, url, ts,
         |  count(ts) over(partition by user, url order by ts range between interval '1' minutes preceding and current row) url_count
         |from click_table
         |""".stripMargin)

    /**
     * Please use window table-valued function with the following computations:
     * 1. aggregate using window_start and window_end as group keys.
     * 2. topN using window_start and window_end as partition key.
     * 3. join with join condition contains window starts equality of input tables and window ends equality of input tables.
     */


    result.execute().print()

    env.execute("WindowTest")
  }
}
